package osgi.common.services.job.jms;

import java.util.Date;

import org.quartz.JobExecutionContext;
import org.quartz.JobExecutionException;

import osgi.common.services.job.client.AbstractJobService;

import javax.jms.Connection;
import javax.jms.DeliveryMode;
import javax.jms.Destination;
import javax.jms.MessageProducer;
import javax.jms.Session;
import javax.jms.TextMessage;

import org.apache.activemq.ActiveMQConnection;
import org.apache.activemq.ActiveMQConnectionFactory;


public class AbstractJMSQueueProducer extends AbstractJobService implements ICommonJMSQueueProducer {

	
	
	protected String  jmsUser        = "helder.morais"; 
	protected String  jmsPassword    = "master"; 
	
	protected String  jmsBrokerUrl   = "tcp://94.1.10.201:61616?soTimeout=2000";
	
	protected Destination destination;

	protected int messageCount = 9;

	protected long sleepTime;

	protected boolean verbose = true;

	protected int messageSize = 255;

	protected long timeToLive;

	
	
	
	protected String  jmsDestinationName     = "TOOL.DEFAULT";
	
	protected boolean topic      = false;
	
	protected boolean transacted = true;
	protected boolean persistent = true;

	protected Connection connection = null;	
	
	protected int jmsTimeOut = 2000;	
	protected int sendTimeOut = jmsTimeOut;
	
	
	
	@Override
	public void execute(JobExecutionContext context) throws JobExecutionException {

		Date agora = new Date();
		showParameters();

		System.out.println("AMQProducerJob : begin - "+agora);

		try {
			
			connection = null;
			// Create the connection.
			//ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(jmsUser, jmsPassword, jmsBrokerUrl);
			ActiveMQConnectionFactory connectionFactory = new ActiveMQConnectionFactory(jmsBrokerUrl);
			
			connectionFactory.setCloseTimeout(sendTimeOut);
			connection = connectionFactory.createConnection();
			
			ActiveMQConnection c = (ActiveMQConnection) connection;
			c.setSendTimeout(sendTimeOut);
			
			connection.start();

			// Create the session
			Session session = connection.createSession(transacted, Session.AUTO_ACKNOWLEDGE);
			if (topic) {
				destination = session.createTopic(jmsDestinationName);
			} else {
				destination = session.createQueue(jmsDestinationName);
			}

			// Create the producer.
			MessageProducer producer = session.createProducer(destination);
			if (persistent) {
				producer.setDeliveryMode(DeliveryMode.PERSISTENT);
			} else {
				producer.setDeliveryMode(DeliveryMode.NON_PERSISTENT);
			}
			if (timeToLive != 0) {
				producer.setTimeToLive(timeToLive);
			}

			try {
				// Start sending messages
				sendMessages(session, producer);
				if (transacted) {
					System.out.println("OK [" + this.getClass().getName() + "] Committing " + messageCount + " messages");
					session.commit();
				}
	
			} catch (Exception e) {
			
                e.printStackTrace();
				
				if (transacted) {
					System.out.println("Error [" + this.getClass().getName() + "] Rolling Back " + messageCount + " messages");
					session.rollback();
				}

			}
			
			
			System.out.println("[" + this.getClass().getName() + "] Done.");
			
			System.out.println("[" + this.getClass().getName() + "] Results:\n");
			
//			c.getConnectionStats().dump(new IndentPrinter());


		} catch (Exception e) {
			System.out.println("[" + this.getClass().getName() + "] Caught: " + e);
			e.printStackTrace();
		} finally {
			try {
				connection.close();
			} catch (Throwable ignore) {
			  System.out.println("[" + this.getClass().getName() + "] ignored: " + ignore);
			}
		}


		System.out.println("AMQProducerJob : end - "+agora);
		
		
	}

	
	
	public void showParameters() {
		System.out.println("Connecting to URL: " + jmsBrokerUrl);
		System.out.println("Publishing a Message with size " + messageSize + " to " + (topic ? "topic" : "queue") + ": " + jmsDestinationName);
		System.out.println("Using " + (persistent ? "persistent" : "non-persistent") + " messages");
		System.out.println("Sleeping between publish " + sleepTime + " ms");
		//System.out.println("Running " + parallelThreads + " parallel threads");

		System.out.println("Login : ${this.user}/${this.password} ");
		if (timeToLive != 0) {
			System.out.println("Messages time to live " + timeToLive + " ms");
		}
	}
	
	
	
	protected void sendMessages(Session session, MessageProducer producer) throws Exception {

		for (int i = 0; i < messageCount || messageCount == 0; i++) {

			TextMessage message = session.createTextMessage(createMessageText(i));

			if (verbose) {
				String msg = message.getText();
				if (msg.length() > 50) {
					msg = msg.substring(0, 50) + "...";
				}
				System.out.println("[" + this.getClass().getName() + "] Sending message: '" + msg + "'");
			}

			producer.send(message);

			if( (i % 5) == 0 ){
				//throw new RuntimeException(" DEU PAUUUU !");
				System.out.println("[" + this.getClass().getName() + "] Begin Sleeping: Time to disconnect ...");
				Thread.currentThread().sleep(5000);
				System.out.println("[" + this.getClass().getName() + "] End Sleeping.");
			}
			
			Thread.sleep(sleepTime);
		}
	}

	protected String createMessageText(int index) {
		StringBuffer buffer = new StringBuffer(messageSize);
		buffer.append("Message: " + index + " sent at: " + new Date());
		if (buffer.length() > messageSize) {
			return buffer.substring(0, messageSize);
		}
		for (int i = buffer.length(); i < messageSize; i++) {
			buffer.append(' ');
		}
		return buffer.toString();
	}



	public String getJmsUser() {
		return jmsUser;
	}



	public void setJmsUser(String jmsUser) {
		this.jmsUser = jmsUser;
	}



	public String getJmsPassword() {
		return jmsPassword;
	}



	public void setJmsPassword(String jmsPassword) {
		this.jmsPassword = jmsPassword;
	}



	public String getJmsBrokerUrl() {
		return jmsBrokerUrl;
	}



	public void setJmsBrokerUrl(String jmsBrokerUrl) {
		this.jmsBrokerUrl = jmsBrokerUrl;
	}



	public boolean isVerbose() {
		return verbose;
	}



	public void setVerbose(boolean verbose) {
		this.verbose = verbose;
	}



	public boolean isTransacted() {
		return transacted;
	}



	public void setTransacted(boolean transacted) {
		this.transacted = transacted;
	}



	public boolean isPersistent() {
		return persistent;
	}



	public void setPersistent(boolean persistent) {
		this.persistent = persistent;
	}



	public String getJmsDestinationName() {
		return jmsDestinationName;
	}



	public void setJmsDestinationName(String jmsDestinationName) {
		this.jmsDestinationName = jmsDestinationName;
	}



	public int getSendTimeOut() {
		return sendTimeOut;
	}



	public void setSendTimeOut(int sendTimeOut) {
		this.sendTimeOut = sendTimeOut;
	}



	public int getJmsTimeOut() {
		return jmsTimeOut;
	}



	public void setJmsTimeOut(int jmsTimeOut) {
		this.jmsTimeOut = jmsTimeOut;
	}	
	
}
